package com.lagou.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.*;

public class HbaseDemo {
    Configuration conf=null;
    Connection conn=null;

    String TABLE_NAME = "user_relation";
    String FAMILY = "friends";

    @Before
    public void init () throws IOException {
        conf = HBaseConfiguration.create();
        // 配置zookeeper
        conf.set("hbase.zookeeper.quorum","hadoop01");
        conf.set("hbase.zookeeper.property.clientPort","2181");
        conn = ConnectionFactory.createConnection(conf);
    }
    // 释放资源
    @After
    public void realse(){
        if(conn != null){
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    // 创建表
    @Test
    public void createTbale() throws IOException {
        HBaseAdmin admin = (HBaseAdmin)conn.getAdmin();
        TableName user_relation = TableName.valueOf(TABLE_NAME);
        // 表不存在在创建
        if(!admin.tableExists(user_relation)){
            // 创建一个表描述对象
            HTableDescriptor descriptor = new HTableDescriptor(user_relation);
            // 指定列族
            descriptor.addFamily(new HColumnDescriptor(FAMILY));
            admin.createTable(descriptor);
        }

        admin.close();
    }

    // 添加用户,准备两个用户相互添加好友和插入数据
    @Test
    public void putUser() throws IOException {
        // 获取表对象
        Table table = conn.getTable(TableName.valueOf(TABLE_NAME));
        // 通过配置文件插入用户
        Properties prop = new Properties();
        prop.load(HbaseDemo.class.getClassLoader().getResourceAsStream("relation.properties"));

        Set<Object> keySet = prop.keySet();
        List<String> listKey = new ArrayList<>();
        for (Object key : keySet) {
            listKey.add(key.toString());
        }
        for (Object uid : keySet) {
            String friendLine = prop.getProperty(uid.toString());
            String[] friends = friendLine.split(",");
            List<Put> listPut = new ArrayList<>();
            for (String friend : friends) {
                if(listKey.contains(friend)){
                    Put put1 = new Put(Bytes.toBytes(friend));
                    put1.addColumn(Bytes.toBytes(FAMILY),Bytes.toBytes(uid.toString()),Bytes.toBytes(uid.toString()));
                    listPut.add(put1);

                    Put put2 = new Put(Bytes.toBytes(uid.toString()));
                    put2.addColumn(Bytes.toBytes(FAMILY),Bytes.toBytes(friend),Bytes.toBytes(friend));
                    listPut.add(put2);
                }
            }

            // 插入数据
            if(!listPut.isEmpty()){
                table.put(listPut);
            }

            listKey.remove(uid.toString());
        }

        table.close();
    }

    // 删除  好友
    @Test
    public void deleteUser() throws IOException {
        // 获取表对象
        Table table = conn.getTable(TableName.valueOf(TABLE_NAME));
        String uid1 = "uid1"; // 用户1
        String uid2 = "uid2"; // 用户1要删除的好友

        // 准备数据
        List<Delete> deletes = new ArrayList();
        Delete delete1 = new Delete(Bytes.toBytes(uid1));
        delete1.addColumn(Bytes.toBytes(FAMILY),Bytes.toBytes(uid2));
        deletes.add(delete1);

        Delete delete2 = new Delete(Bytes.toBytes(uid2));
        delete2.addColumn(Bytes.toBytes(FAMILY),Bytes.toBytes(uid1));
        deletes.add(delete2);

        // 执行删除
        table.delete(deletes);

        table.close();
    }
}
